package assign2;

import java.util.concurrent.Semaphore;


public class assign2_q16b {

	final int numConsumers = 4;
	final int bufferSize = 5;
	MultiConsumptionBuffer multiConsumeBuffer = new MultiConsumptionBuffer();
	Consumer[] consumerArray = new Consumer[numConsumers];
	
	public assign2_q16b()
	{
    	for(int i = 0; i<numConsumers; ++i)
    	{
    		consumerArray[i] = new Consumer(i);
    		consumerArray[i].start();
    	}
    	
		Producer producer = new Producer(0);
		producer.start();
	}
	
	
    public class Producer extends Thread
    {
    	// Invariant: 0 <= _rear < bufferSize
    	private int _rear = 0;
    	private int _num;
    	
    	public Producer(int num)
    	{
    		_num = num;
    	}
    	
    	public void run()
    	{
    		System.out.println("run producer");
    		while(true)
    		{
    			try 
    			{ 
    				multiConsumeBuffer.empty.acquire();
    			} 
    			catch (InterruptedException e1) {}
				
    			int product = (int)(Math.random()*1000.0);
    			System.out.println("producer" + _num + ":" + product);
    			
    			try 
    			{ 
    				multiConsumeBuffer.slots[_rear].slotLock.acquire(); // ENTER critical section on slots[_rear].slotLock
    			} 
    			catch (InterruptedException e1) {}
				
    			multiConsumeBuffer.slots[_rear].deposit(product);
    			multiConsumeBuffer.slots[_rear].slotLock.release(); // EXIT critical section on slots[_rear].slotLock
    			_rear = (_rear + 1) % bufferSize;
    			
    			try{Thread.sleep((long)100);}
				catch (InterruptedException e){}
    		}
    	}
    }
    
    public class Consumer extends Thread
    {
    	// have each consumer maintain it's own front location to allow
    	// receiving of messages at different times
    	
    	// Invariant: 0 <= _front < bufferSize
    	private int _front = 0;
    	private int _num;
    	public Semaphore _full = new Semaphore(0);
    	
    	public Consumer(int num)
    	{
    		_num = num;
    	}
    	
    	public void run()
    	{
    		System.out.println("run consumer");
    		while(true)
    		{
    			try
    			{
    				_full.acquire();
    			}
    			catch(InterruptedException e1) {}
    			
    			int product;
    			
    			try
    			{
    				multiConsumeBuffer.slots[_front].slotLock.acquire(); // ENTER critical section on slots[_front].slotLock
    			}
    			catch(InterruptedException e1) {}
    			
    			product = multiConsumeBuffer.slots[_front].consume();
    			multiConsumeBuffer.slots[_front].slotLock.release(); // ENTER critical section on slots[_front].slotLock
    			_front = (_front + 1) % bufferSize;
    			System.out.println("consumer"+ _num + ":" + product);
    			
    			try{Thread.sleep((long)(Math.random() * 3000) + 100);}
				catch (InterruptedException e){}
    		}
    	}
    }
    
    /** This buffer and its associated subclass implement a multiple consumer capable
     *  buffer system
     * 
     * Invariants:
     * -> empty = Number of slots which have been totally consumed by 'numConsumers' number of consumers /\
     *    slots.slotLock = 0 || 1 (intended to act as binary semaphore) /\
     *    0 <= slots._consumptionCount <= numConsumers
     */
	public class MultiConsumptionBuffer {

		public MultiConsumptionSlot[] slots = new MultiConsumptionSlot[bufferSize];
		public Semaphore empty = new Semaphore(bufferSize);
		
		public MultiConsumptionBuffer(){
	    	for(int i = 0; i < bufferSize; ++i)
	    	{
	    		slots[i] = new MultiConsumptionSlot(0, numConsumers);
	    		System.out.println("createing multi consumption slot: " + i);
	    	}
		}
		
		
		public class MultiConsumptionSlot {
			private int _consumptionsRequired;
			private int _consumptionCount = 0;
			private int _message;
			// all producers or consumers must acquire this lock prior to operating
			// on the slot and release it afterwards
			public Semaphore slotLock = new Semaphore(1);

			public MultiConsumptionSlot(int message, int consumptionsRequired) {
				_consumptionsRequired = consumptionsRequired;
				_message = message;
			}

			int consume() {
				_consumptionCount++;
				if (_consumptionCount == _consumptionsRequired) {
					
					// Signal the buffer to grant and empty permit
					empty.release();
				}
				return _message;
			}

			void deposit(int message) {
				// assert _consumptionCount = _consumptionsRequired at this point
				_message = message;
				_consumptionCount = 0;

				// give all consumers a full permit
				for (int i = 0; i < numConsumers; i++) {
					consumerArray[i]._full.release();
				}
			}
		}

	}

	public static void main(String[] args){
		new assign2_q16b();
	}
}
